package redis.client.nedis;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import redis.client.nedis.commands.RedisCommand;
import redis.client.nedis.commands.ServerCommands;
import redis.client.nedis.reply.IReply;
import redis.client.nedis.request.RedisRequest;
import redis.client.nedis.request.RedisRequestQueue;
import redis.client.nedis.util.RedisRequestUtil;

import java.util.function.Consumer;

/**
 * redis客户端
 *
 * @author gaodong
 */
public class Client {

    final EventLoopGroup bossGroup = new NioEventLoopGroup(Runtime.getRuntime().availableProcessors());
    final Bootstrap bootstrap = new Bootstrap();

    private Channel channel;
    private ChannelFuture channelFuture;
    private HostAndPort hostAndPort;
    private String auth;

    public Client(String ip, int port, String auth) {
        this.auth = auth;
        hostAndPort = HostAndPort.build(ip, port);
        initBootstrap();
        checkServer();
    }

    public void initBootstrap() {
        bootstrap.group(bossGroup)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.TCP_NODELAY, true)
                .option(ChannelOption.SO_KEEPALIVE, true)
                .remoteAddress(hostAndPort.getHost(), hostAndPort.getPort())
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel socketChannel) {
                        ChannelPipeline pl = socketChannel.pipeline();
                        pl.addLast(new RedisCommandEncoder());
                        pl.addLast(new RedisReplyDecoder());
                        pl.addLast(new ReceiveClientHandler(hostAndPort));
                    }
                });
    }

    public Channel refreshChannel() {
        if (channel == null) {
            try {
                channel = bootstrap.connect().sync().channel();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        return channel;
    }

    public ChannelFuture refreshChannelFuture() {
        if (channel == null) {
            try {
                channelFuture = bootstrap.connect().sync();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        return channelFuture;
    }

    /**
     * 验证redis性质，是否集群等
     */
    private void checkServer() {

        refreshChannel();

        // 密码验证
        if (auth != null && !auth.isEmpty()) {
            try {
                RedisCommand rc = RedisCommand.getInstance(ServerCommands.AUTH, auth);
                RedisRequest rq = new RedisRequest();
                rq.setHostAndPort(hostAndPort);
                rq.setCommand(rc);
                RedisRequestUtil.sendRequests(channel, rq);
                rq.getSemaphore().acquire();
                IReply<?> rs = rq.getReply();
                if ("OK".equals(rs.data())) {
                    System.out.println("auth passed!");
                } else {
                    System.out.println(rs);
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

        RedisRequestQueue.getInstance().init(hostAndPort);
    }

    /**
     * 关闭客户端
     */
    public void close() {
        bossGroup.shutdownGracefully();
    }

    public IReply<?> get(String key, Consumer<RedisRequest> customer) {
        RedisCommand rc = RedisCommand.getInstance(ServerCommands.GET, key);
        return invoke(customer, rc)[0];
    }

    public IReply<?> set(String key, String value, Consumer<RedisRequest> customer) {
        RedisCommand rc = RedisCommand.getInstance(ServerCommands.SET, key, value);
        return invoke(customer, rc)[0];
    }

    public IReply<?>[] testpipe(Consumer<RedisRequest> consumer) {
        RedisCommand[] rcs = new RedisCommand[1000];
        for (int i = 0; i < rcs.length; i++) {
            rcs[i] = RedisCommand.getInstance(ServerCommands.SET, "key" + i, String.valueOf(i));
        }
        return invoke(consumer, rcs);
    }

    private IReply<?>[] invoke(Consumer<RedisRequest> consumer, RedisCommand... rcs) {
        RedisRequest[] rqs = new RedisRequest[rcs.length];
        int i = 0;
        for (RedisCommand rc : rcs) {
            rqs[i++] = new RedisRequest(rc, hostAndPort, consumer);
        }
        return invoke(rqs);
    }

    private IReply<?>[] invoke(RedisRequest... requests) {

        long t1 = System.nanoTime();

        refreshChannel();

        long t2 = System.nanoTime();

        RedisRequestUtil.sendRequests(channel, requests);

        long t3 = System.nanoTime();

        /*for (RedisRequest rq : requests) {
            rq.acquireMutex();
        }*/
       /* for (; ; ) {
            boolean finished = true;
            for (RedisRequest rq : requests) {
                if (rq.getReply() == null) {
                    finished = false;
                    break;
                }
            }
            if (finished) {
                break;
            }
            try {
                Thread.sleep(0);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }*/

        long t4 = System.nanoTime();

        IReply<?>[] objs = new IReply<?>[requests.length];
        int i = 0;
        for (RedisRequest rq : requests) {
            objs[i++] = rq.getReply();
        }

        long t5 = System.nanoTime();

        System.out.println("t1=" + (t2 - t1) / 1e6 + "ms");
        System.out.println("t2=" + (t3 - t2) / 1e6 + "ms");
        System.out.println("t3=" + (t4 - t3) / 1e6 + "ms");
        System.out.println("t4=" + (t5 - t4) / 1e6 + "ms");

        return objs;
    }
}
